package bma.common.thrift;

import java.util.concurrent.ThreadFactory;

import org.apache.thrift.TProcessor;
import org.apache.thrift.server.TNonblockingServer;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TSimpleServer;
import org.apache.thrift.server.TThreadPoolServer;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingServerTransport;
import org.apache.thrift.transport.TServerTransport;
import org.apache.thrift.transport.TTransportFactory;

import bma.common.langutil.core.Preconditions;

public abstract class ThriftServer {

	final org.slf4j.Logger log = org.slf4j.LoggerFactory
			.getLogger(ThriftServer.class);

	private TServer server;

	public ThriftServer() {
		super();
	}

	public ThriftServer(TServer server) {
		super();
		this.server = server;
	}

	public TServer getServer() {
		return server;
	}

	public void setServer(TServer server) {
		this.server = server;
	}

	public abstract void start();

	public void close() {
		if (this.server != null) {
			this.server.stop();
		}
	}

	public static class NewThreadServer extends ThriftServer {

		private ThreadFactory threadFactory;

		public NewThreadServer() {
			super();
		}

		public NewThreadServer(TServer server) {
			super(server);
		}

		public ThreadFactory getThreadFactory() {
			return threadFactory;
		}

		public void setThreadFactory(ThreadFactory threadFactory) {
			this.threadFactory = threadFactory;
		}

		public void start() {
			Preconditions.checkNotNull(getServer());
			Runnable r = new Runnable() {

				@Override
				public void run() {
					try {
						getServer().serve();
					} catch (Exception e) {
						log.warn(getServer() + " serve fail", e);
					}
				}
			};

			Thread th = null;
			if (this.threadFactory != null) {
				th = this.threadFactory.newThread(r);
			} else {
				th = new Thread(r);
			}
			th.start();
		}
	}

	public static class SimpleServer extends NewThreadServer {

		public SimpleServer(TServerTransport trans, TProcessor processor) {
			super();
			TProcessor p = processor;
			if (!(p instanceof TProcessor4Error)) {
				p = new TProcessorWrap(processor);
			}
			setServer(new TSimpleServer(new TServer.Args(trans).processor(p)));
		}

	}

	public static class ThreadPoolServer extends NewThreadServer {

		public ThreadPoolServer(TServerTransport trans, TProcessor processor,
				int maxWorker, int minWorker) {
			super();
			TProcessor p = processor;
			if (!(p instanceof TProcessor4Error)) {
				p = new TProcessorWrap(processor);
			}
			setServer(new TThreadPoolServer(new TThreadPoolServer.Args(trans)
					.processor(p).maxWorkerThreads(maxWorker)
					.minWorkerThreads(minWorker)));
		}

	}

	public static class HsHaServer extends NewThreadServer {

		public HsHaServer(TNonblockingServerTransport trans,
				TProcessor processor, int worker, int maxBufferSize) {
			super();
			TProcessor p = processor;
			if (!(p instanceof TProcessor4Error)) {
				p = new TProcessorWrap(processor);
			}
			TTransportFactory fac = new TFramedTransport.Factory(maxBufferSize);
			// setServer(new THsHaServer(new
			// THsHaServer.Args(trans).processor(p)
			// .workerThreads(worker).inputTransportFactory(fac)
			// .outputTransportFactory(fac)));
			TNonblockingServer.Args args = new TNonblockingServer.Args(trans)
					.processor(p).inputTransportFactory(fac)
					.outputTransportFactory(fac);
			args.maxReadBufferBytes = maxBufferSize;
			setServer(new TNonblockingServer(args));
		}
	}
}
